package com.zy.datareceiver.binsess.rabbit.service;


import com.rabbitmq.client.Channel;
import com.zy.datareceiver.binsess.rabbit.entity.DataGram;
import com.zy.datareceiver.binsess.rabbit.entity.Head;
import com.zy.datareceiver.binsess.rabbit.entity.PassagewayDomain;
import com.zy.datareceiver.binsess.rabbit.entity.ReceiveDataStorage;
import com.zy.datareceiver.core.cast.CacheDict;
import com.zy.datareceiver.core.cast.ConCast;
import com.zy.datareceiver.core.utils.JsonUtil;
import org.codehaus.jackson.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * <p>
 *     @See {@link RabbitMqTaskReceiver#monitoringDataProcess(byte[], Channel, Map)}
 *     数据接收，直接从rabbitmq中的topicAll取出数据，数据接收采用字节数组进行接收
 * </p>
 * <p>
 *     整体流程：首先接受到数据之后将数据转换成字符串形式，然后将数据（阻塞存放，
 *     队列中数据满了线程阻塞）存放在内部维护的暂存队列，ack确认，告诉rabbitmq
 *     目前序号为deliveryTag的这一条数据已经收到可以删除了
 * </p>
 * <p>
 *     @See {@link RabbitMqTaskReceiver#stateProcess(byte[], Channel, Map)}
 *     是传感器状态数据的接收方法，从rabbitmq中的topicState中取数据,同理会先放入暂存队列中，然后在进行ack确认
 * </p>
 * */
@Component
public class RabbitMqTaskReceiver {


    private static final Logger log = LoggerFactory.getLogger(RabbitMqTaskReceiver.class);


    @RabbitListener(queues = ConCast.Topic.MONITORING_DATA)
    @RabbitHandler
    public void monitoringDataProcess(@Payload byte[] body, Channel channel, @Headers Map<String,Object> prop)
            throws IOException, InterruptedException , TimeoutException {

        String msg = new String(body,"UTF8");
        log.info("receiver data {}",msg);

        putMessage2ReceiveQueue(msg);
        callbackAck(channel,prop,msg);
    }

    @RabbitListener(queues = ConCast.Topic.STATUS_DATA)
    @RabbitHandler
    public void stateProcess(@Payload byte[] body, Channel channel, @Headers Map<String,Object> prop)
            throws IOException, InterruptedException ,TimeoutException{

        String msg = new String(body,"UTF8");
        log.info("收到心跳数据包 {}",msg);
        PassagewayDomain passagewayDomain = JsonUtil.string2Object(msg, new TypeReference<PassagewayDomain>() {});

        ReceiveDataStorage.getInstance().statusQueue.offer(passagewayDomain);
        callbackAck(channel,prop,msg);

    }

    private void callbackAck(Channel channel, Map<String, Object> prop,String msg) {
        Long deliveryTag = null;
        try {
            deliveryTag= (Long)prop.get(AmqpHeaders.DELIVERY_TAG);
            //手工ACK,告诉rabbitmq，目前已经收到id为deliveryTag的消息，你可以删除了
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            //丢弃这条消息
            try {
                //最后一个参数是：是否重回队列
                channel.basicNack(deliveryTag, false,false);
            } catch (IOException e1) {
                log.error("ACK 处理失败：{}",deliveryTag,msg);
            }
            log.debug("消息消费失败：id：{}",deliveryTag,msg);
        }
    }



    /**
     * 如果pointId为空，直接定义为异常数据，直接返回，不为空则
     * 将数据存放到中间队列中，在存放的时候，如果队列数据已满，线程阻塞
     * */
    private void putMessage2ReceiveQueue(String msg) {
        DataGram dataGram = JsonUtil.string2Object(msg, new TypeReference<DataGram>() {});
        Integer pointId = dataGram.getHead().getPointId();
        if (pointId == null){
            log.info("异常数据 {}",JsonUtil.obj2String(dataGram));
            return;
        }
        DataGram queueDataGram  = fillDataGram(dataGram);
        try {
            ReceiveDataStorage.getInstance().receiveQueue.put(queueDataGram);
        } catch (InterruptedException e) {
            log.error("本地队列已满，往本地数据队列投放失败 data: {}",JsonUtil.obj2String(dataGram));
        }
    }

    /**
     * 对dataGram的数据进行填充，会存在pointName或者deviceName等信息不全的情况，
     * 为了便于后续处理，于是在这里根据poinId进行数据的填充，统一返回，然后将填充
     * 之后的数据放入队列中，保证数据的完整性
     * */
    private DataGram fillDataGram(DataGram dataGram) {
        String pointName = dataGram.getHead().getPointName();
        String deviceName = dataGram.getHead().getDeviceName();
        Integer deviceId = dataGram.getHead().getDeviceId();
        if (pointName == null || deviceId == null || deviceName == null){
            Integer pointId = dataGram.getHead().getPointId();
            Head head = dataGram.getHead();
            head.setPointName(CacheDict.getPointNameByPointId(pointId));
            head.setDeviceName(CacheDict.getDeviceNameByPointId(pointId));
            head.setDeviceId(CacheDict.getDeviceIdByPointId(pointId));
            return dataGram;
        }
        return dataGram;
    }
}
